packages/core/src/core/db/mongo-db.ts (1)

Line range hint 212-239: Unsuccessful queries if _id is stored as ObjectId.

The code uses { _id: orchestratorId } but orchestratorId is a string. If the _id in MongoDB is an ObjectId, this query will fail to match. Convert the string to an ObjectId or store _id as a string consistently:

await this.orchestratorCollection.updateOne(
-  { _id: orchestratorId },
+  { _id: new ObjectId(orchestratorId) },
  ...
);

...

return this.orchestratorCollection.findOne({
-  _id: orchestratorId,
+  _id: new ObjectId(orchestratorId),
});

Also applies to: 261-265, 268-283

packages/core/src/core/orchestrator.ts (1)

Line range hint 258-282: Ensure getOrchestratorById returns the correct record.

Because _id in Mongo might be stored as an ObjectId while orchestratorId is a string, this call risks returning null. Consider either storing _id as a string or casting from string to ObjectId in the query. Otherwise, each new orchestrator attempt will create a new record unnecessarily.

examples/example-server.ts (1)

Line range hint 228-252: Inconsistent ID handling in API endpoints.

The API endpoints still use ObjectId while the rest of the codebase has moved to string IDs. This inconsistency should be addressed.

-        let objectId;
-        try {
-            objectId = new ObjectId(chatId);
-        } catch (err) {
-            return res.status(400).json({ error: "Invalid chat ID format" });
-        }
-
-        const history = await scheduledTaskDb.getOrchestratorById(objectId);
+        const history = await scheduledTaskDb.getOrchestratorById(chatId);

packages/core/src/core/schedule-service.ts (3)

23-32: Consider concurrency checks or leadership election for multiple service instances.

If multiple SchedulerService instances run in parallel, there is a possibility that more than one instance may pick up the same tasks. To avoid duplicate processing, consider implementing a “leadership election” or adding a stronger guard (like an atomic “compare and set” on task status) to ensure that only one scheduler processes a pending task at a time.

---

90-119: Validate scheduling parameters to prevent misconfigured intervals.

If an end-user sets a negative or extremely large interval, it could result in scheduling anomalies. A quick validation step (e.g., ensuring intervalMs >= 0) helps avoid unexpected behavior.

---

120-125: Consider graceful shutdown.

Although stop() clears the polling interval, any in-flight tasks won’t be canceled. If you need a truly graceful shutdown, ensure you also wait for currently running tasks to finish before exiting your application.

packages/core/src/core/db/mongo-db.ts (2)

133-133: Use a "failed" status for errors.
Marking tasks as “completed” or “failed” is a good approach to categorize the final outcome. Make sure that any relevant logs or metrics reflect the difference so that failed tasks can be retried or inspected.

---

150-150: Revert the status to “pending” only after verifying concurrency.

If multiple service instances run, one instance could update the same task’s nextRunAt, while another instance is concurrently running it. Consider an additional check to confirm the current status before re-pending the task.

packages/core/src/core/orchestrator.ts (1)

356-379: Scheduling tasks here is consistent but verify concurrency approach.

Creating tasks directly in the Orchestrator helps chain further actions. However, if multiple Orchestrator instances run concurrently, tasks may be scheduled more than once. Consider a uniqueness clause or deduplication approach to avoid duplicates for the same user/event.

packages/core/src/core/memory.ts (2)

33-34: Consider adding error types for database operations.

The connect and close methods could benefit from specific error types to handle different failure scenarios.

-    connect(): Promise<void>;
-    close(): Promise<void>;
+    connect(): Promise<void | DatabaseConnectionError>;
+    close(): Promise<void | DatabaseCloseError>;

---

9-9: Consider using a more specific type for taskData.

Using Record<string, any> reduces type safety. Consider defining a specific interface or type for task data based on your use cases.

-    taskData: Record<string, any>;
+    taskData: TaskData;

+interface TaskData {
+    // Define specific fields based on your use cases
+    [key: string]: string | number | boolean | object;
+}

examples/example-twitter.ts (1)

179-185: Consider extracting magic numbers for intervals.

The interval values should be defined as named constants at the top of the file for better maintainability.

+const MENTION_CHECK_INTERVAL_MS = 6000;  // Check mentions every minute
+const CONSCIOUSNESS_INTERVAL_MS = 30000;  // Think every 5 minutes

-    await scheduler.scheduleTaskInDb("sleever", "twitter_mentions", {}, 6000);
-    await scheduler.scheduleTaskInDb("sleever", "consciousness_thoughts", {}, 30000);
+    await scheduler.scheduleTaskInDb("sleever", "twitter_mentions", {}, MENTION_CHECK_INTERVAL_MS);
+    await scheduler.scheduleTaskInDb("sleever", "consciousness_thoughts", {}, CONSCIOUSNESS_INTERVAL_MS);

examples/example-api.ts (1)

Line range hint 1-13: Update file documentation to reflect architectural changes.

The file header comments should be updated to:

1. Remove Twitter-specific references since this is now a general API example
2. Document the database abstraction layer usage

Apply this diff to update the documentation:

 /**
- * Example demonstrating a Twitter bot using the Daydreams package.
+ * Example demonstrating a general API integration using the Daydreams package.
  * This bot can:
- * - Monitor Twitter mentions and auto-reply
- * - Generate autonomous thoughts and tweet them
+ * - Process API requests and auto-reply
+ * - Generate autonomous thoughts
  * - Maintain conversation memory using ChromaDB
+ * - Store scheduled tasks using MongoDB
  * - Process inputs through a character-based personality
  */

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

packages/core/src/core/processors/master-processor.ts (4)

17-33: Use a constructor-level comment or docstring to clarify usage and dependencies.

The constructor’s parameters are self-explanatory but adding a short docstring clarifying how the character, llmClient, and other fields fit together would improve maintainability and onboarding.

---

39-46: Ensure alignment of canHandle logic with contentLimit.

Currently, canHandle checks if the content length is < 1000, matching the default in the base class constructor. If you allow further changes to contentLimit, consider referencing this.contentLimit explicitly in the docstring or method name for clarity.

---

215-221: Leverage optional chaining to reduce complexity.

Static analysis suggests using an optional chain for safety. However, your current check is still valid. If you prefer a fluent style, consider this refactor:

- if (childProcessor && childProcessor.canHandle(content)) {
+ if (childProcessor?.canHandle(content)) {

🧰 Tools

🪛 Biome (1.9.4)

[error] 221-221: Change to an optional chain.

Unsafe fix: Change to an optional chain.

(lint/complexity/useOptionalChain)

---

238-285: Provide more nuanced fallback handling.

When an error occurs, the fallback response always returns neutral sentiment without giving the user enough context. Consider capturing partial results (if any) or providing an error field to alert upstream consumers.

packages/core/src/core/processor.ts (5)

10-11: Encourage a typed record for child processors.

While Map<string, BaseProcessor> is adequate, using a typed object literal or an expanded interface might enable clearer relationships and reduce possible runtime errors.

---

15-17: Constructor parameters could benefit from partial typing or defaults.

You removed the default for loggerLevel. If there are no strict requirements that it must always be provided, consider a default to avoid collisions in other places that instantiate BaseProcessor.

---

34-39: Expand docstring to convey the intended usage of getDescription.

This method currently returns the processor’s description. Documenting when and how it might be used (e.g., debugging, generating help text) helps future maintainers.

---

58-72: Validate child processors’ compatibility prior to adding.

addProcessor throws an error if a processor name exists. That’s a good safeguard. Additionally, consider verifying that child processors have unique roles or are valid for delegation, so you avoid hidden conflicts in the chain of responsibility.

---

74-79: Provide fallback or log if the requested name does not exist.

When getProcessor is called with a name that does not exist, returning undefined might silence debugging signals. Logging an info or debug message could help trace errors faster in large deployments.

packages/core/src/core/orchestrator.ts (2)

218-225: Move debug logging before action execution.

The debug logging should be placed before executing the action to ensure we capture the input context even if the action fails.

     try {
+        this.logger.debug(
+            "Orchestrator.dispatchToAction",
+            "Executing action",
+            {
+                name,
+                data,
+            }
+        );
         const result = await handler.execute(data);
-        this.logger.debug(
-            "Orchestrator.dispatchToAction",
-            "Executing action",
-            {
-                name,
-                data,
-            }
-        );
         return result;

---

346-369: Extract task scheduling logic into a separate method.

The task scheduling logic could be extracted into a dedicated method for better maintainability and reusability.

+ private async scheduleTask(
+     userId: string,
+     task: { name: string; data: any; intervalMs?: number }
+ ): Promise<void> {
+     const now = Date.now();
+     const nextRunAt = new Date(now + (task.intervalMs ?? 0));
+
+     this.logger.info(
+         "Orchestrator.scheduleTask",
+         `Scheduling task ${task.name}`,
+         {
+             nextRunAt,
+             intervalMs: task.intervalMs,
+         }
+     );
+
+     await this.orchestratorDb.createTask(
+         userId,
+         task.name,
+         {
+             request: task.name,
+             task_data: JSON.stringify(task.data),
+         },
+         nextRunAt,
+         task.intervalMs
+     );
+ }

  // In runAutonomousFlow:
- const now = Date.now();
- const nextRunAt = new Date(now + (task.intervalMs ?? 0));
-
- this.logger.info(
-     "Orchestrator.runAutonomousFlow",
-     `Scheduling task ${task.name}`,
-     {
-         nextRunAt,
-         intervalMs: task.intervalMs,
-     }
- );
-
- await this.orchestratorDb.createTask(
-     userId,
-     task.name,
-     {
-         request: task.name,
-         task_data: JSON.stringify(task.data),
-     },
-     nextRunAt,
-     task.intervalMs
- );
+ await this.scheduleTask(userId, task);

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

packages/core/src/core/processors/chain-of-thought-processor.ts (3)

8-31: Encapsulate or validate memory usage for reliable DB abstraction.
The constructor injects memory: VectorDB directly into the ChainOfThought instance. While this is consistent with the new DB abstraction approach, consider adding validation or error handling if the injected VectorDB fails to initialize or has limited availability (e.g., unavailable network resources). Encapsulating these checks can improve the processor’s resilience and clarify the system’s resource requirements.

---

33-49: Validate textual content more robustly in canHandle.
The canHandle method checks for keywords ("goal", "plan", etc.) and content length but does not account for edge cases like extremely large or non-textual input (e.g., binary data, HTML). For broader usage, consider implementing defensive checks or limiting memory usage for large inputs.

---

211-218: Extend error information when rethrowing.
While rethrowing the error preserves the stack trace, consider adding context about which content triggered the failure for extended observability. This helps pinpoint the cause of repeated errors in logs or incident reports.

packages/core/src/core/chain-of-thought.ts (1)

31-31: Reevaluate making stepManager public.
The stepManager property was changed to public, potentially exposing it to outside modifications. If external classes only need read access, consider a getter or read-only property. This preserves encapsulation while still supporting your new scheduler architecture and DB abstraction.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

packages/core/src/core/orchestrator.ts (1)

Line range hint 502-529: Improve error handling in dispatchToInput.

The error handling could be more informative and consistent with the rest of the codebase:

 try {
     const result = await handler.execute(data);
     if (result) {
         return await this.runAutonomousFlow(
             result,
             handler.name,
             userId,
             orchestratorId
         );
     }
     return [];
 } catch (error) {
-    this.logger.error(
-        "dispatchToInput Error",
-        `dispatchToInput Error: ${
-            error instanceof Error ? error.message : String(error)
-        }`
-    );
+    this.logger.error(
+        "Orchestrator.dispatchToInput",
+        "Failed to execute input handler",
+        {
+            name,
+            error: error instanceof Error ? error.message : String(error),
+            userId,
+            orchestratorId
+        }
+    );
+    throw error; // Re-throw to allow proper error handling upstream
 }

packages/core/src/core/processors/master-processor.ts (4)

17-33: Consider enhancing logging and description for better observability.

As this is a master processor responsible for content delegation:

1. Consider using LogLevel.INFO as the default to improve observability of routing decisions
2. Enhance the description to better reflect its role in content delegation and processing orchestration

-        logLevel: LogLevel = LogLevel.ERROR
+        logLevel: LogLevel = LogLevel.INFO
         ) {
         super(
             {
                 name: "master",
                 description:
-                    "This processor handles messages or short text inputs.",
+                    "Master processor that analyzes content, delegates to specialized processors, and orchestrates processing flows for messages and short text inputs.",
             },

---

48-115: Extract prompt construction to improve maintainability.

The prompt construction logic is complex and could benefit from being moved to a separate method or configuration file. This would make it easier to maintain and test the prompt template independently.

+    protected getProcessingPrompt(
+        content: string,
+        otherContext: string,
+        processorContext: string,
+        outputsSchemaPart?: string,
+        actionsSchemaPart?: string
+    ): string {
+        return `You are a master processor...`; // Move the entire prompt template here
+    }
+
     async process(
         content: any,
         otherContext: string,
         ioContext?: {
             availableOutputs: OutputIOHandler[];
             availableActions: ActionIOHandler[];
         }
     ): Promise<ProcessedResult> {

---

199-219: Use optional chaining as suggested by static analysis.

The code could benefit from using optional chaining for better readability and safety.

-            if (result.classification.delegateToProcessor) {
-                const childProcessor = this.getProcessor(
-                    result.classification.delegateToProcessor
-                );
-                if (childProcessor && childProcessor.canHandle(content)) {
+            const childProcessor = result.classification.delegateToProcessor
+                ? this.getProcessor(result.classification.delegateToProcessor)
+                : null;
+            if (childProcessor?.canHandle(content)) {

🧰 Tools

🪛 Biome (1.9.4)

[error] 204-204: Change to an optional chain.

Unsafe fix: Change to an optional chain.

(lint/complexity/useOptionalChain)

---

246-248: Enhance error logging for better debugging.

The error logging could be more detailed to help with debugging processing failures.

-            this.logger.error("Processor.process", "Processing failed", {
-                error,
-            });
+            this.logger.error("Processor.process", "Processing failed", {
+                error,
+                content: typeof content === 'string' ? content : JSON.stringify(content),
+                errorMessage: error instanceof Error ? error.message : String(error),
+                stack: error instanceof Error ? error.stack : undefined
+            });

packages/core/src/core/orchestrator.ts (3)

46-53: Consider reordering constructor parameters for better maintainability.

The constructor parameters mix core dependencies with optional configurations. Consider grouping required dependencies first, followed by optional parameters:

 constructor(
-    private readonly roomManager: RoomManager,
-    vectorDb: VectorDB,
-    private processor: BaseProcessor,
-    orchestratorDb: OrchestratorDb,
-    config?: LoggerConfig
+    private readonly orchestratorDb: OrchestratorDb,
+    private readonly processor: BaseProcessor,
+    private readonly roomManager: RoomManager,
+    vectorDb: VectorDB,
+    config?: LoggerConfig
 )

---

346-369: Refactor task scheduling logic into a separate method.

The task scheduling logic within runAutonomousFlow is complex and would be more maintainable as a separate method.

+private async scheduleTask(
+    userId: string,
+    task: { name: string; intervalMs?: number; data: unknown }
+): Promise<void> {
+    const now = Date.now();
+    const nextRunAt = new Date(now + (task.intervalMs ?? 0));
+
+    this.logger.info(
+        "Orchestrator.scheduleTask",
+        `Scheduling task ${task.name}`,
+        {
+            nextRunAt,
+            intervalMs: task.intervalMs,
+        }
+    );
+
+    await this.orchestratorDb.createTask(
+        userId,
+        task.name,
+        {
+            request: task.name,
+            task_data: JSON.stringify(task.data),
+        },
+        nextRunAt,
+        task.intervalMs
+    );
+}

 // In runAutonomousFlow:
-const now = Date.now();
-const nextRunAt = new Date(now + (task.intervalMs ?? 0));
-
-this.logger.info(
-    "Orchestrator.runAutonomousFlow",
-    `Scheduling task ${task.name}`,
-    {
-        nextRunAt,
-        intervalMs: task.intervalMs,
-    }
-);
-
-await this.orchestratorDb.createTask(
-    userId,
-    task.name,
-    {
-        request: task.name,
-        task_data: JSON.stringify(task.data),
-    },
-    nextRunAt,
-    task.intervalMs
-);
+await this.scheduleTask(userId, task);

---

8-8: Add JSDoc documentation for the OrchestratorDb interface.

The OrchestratorDb interface is a crucial part of the system's architecture, but it lacks proper documentation. Consider adding comprehensive JSDoc comments to document the interface's purpose, methods, and expected behavior.

+/**
+ * Interface for database operations related to orchestrator functionality.
+ * Implementations of this interface handle persistence of orchestrator state,
+ * messages, and scheduled tasks.
+ */
 import type { OrchestratorDb } from "./memory";

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

packages/core/src/core/schedule-service.ts (1)

34-102: ⚠️ Potential issue

Handle task failures with a fallback flow.

If an error occurs after calling markRunning but before calling markCompleted, the task remains stuck in the "running" state.

     try {
         const tasks = await this.context.orchestratorDb.findDueTasks();
         for (const task of tasks) {
+            try {
                 await this.context.orchestratorDb.markRunning(task._id);
                 const handler = this.orchestrator.getHandler(task.handlerName);
                 ...
+            } catch (error) {
+                await this.context.orchestratorDb.markCompleted(task._id, true);
+                this.context.logger.error("Task processing error", "error", { taskId: task._id, error });
+                continue;
+            }
         }
     } catch (err) {
         this.context.logger.error("pollTasks error", "error", { data: err });
     }

packages/core/src/core/orchestrator.ts (1)

290-294: ⚠️ Potential issue

Fix potential race condition in orchestrator creation.

There's a potential race condition between checking for an existing orchestrator and creating a new one.

Consider using a transaction or implementing proper locking:

-const existing =
-    await this.orchestratorDb.getOrchestratorById(orchestratorId);
-if (!existing) {
-    orchestratorId =
-        await this.orchestratorDb.createOrchestrator(userId);
-}
+orchestratorId = await this.orchestratorDb.getOrCreateOrchestrator(
+    orchestratorId,
+    userId
+);

packages/core/src/core/schedule-service.ts (1)

14-32: Add validation for the polling interval.

The polling interval should be validated to ensure it's within reasonable bounds (e.g., not too short to avoid overwhelming the system, not too long to maintain responsiveness).

 constructor(
     private context: IOrchestratorContext,
     private orchestrator: Orchestrator,
-    private pollMs: number = 10_000
+    private pollMs: number = 10_000
 ) {
+    if (pollMs < 1000) {
+        throw new Error('Polling interval must be at least 1 second');
+    }
+    if (pollMs > 3600000) {
+        throw new Error('Polling interval must not exceed 1 hour');
+    }
 }

packages/core/src/core/types/index.ts (2)

288-288: Document the nextProcessor property usage.

The nextProcessor property enables processor chaining, but its usage pattern isn't immediately clear.

Add JSDoc comments to explain:

• Expected format of the processor name
• How it's used in the processing pipeline
• Example usage

+  /**
+   * Name of the next processor to handle this result.
+   * @example "sentiment-analyzer" or "translation-processor"
+   */
   nextProcessor?: string;

---

614-616: Consider extending AgentRequest interface.

The interface currently only includes headers, but might need to be extended in the future.

Consider adding commonly needed request properties:

 export interface AgentRequest {
     headers: Record<string, string>;
+    method?: string;
+    path?: string;
+    query?: Record<string, string>;
+    body?: unknown;
 }

packages/core/src/core/orchestrator.ts (1)

592-600: Consider batching memory operations.

The code performs multiple database operations sequentially when saving memories.

Consider batching the operations for better performance:

-await this.roomManager.addMemory(
-    content.room,
-    JSON.stringify(result.content),
-    {
-        source,
-        ...result.metadata,
-        ...result.enrichedContext,
-    }
-);
-await this.roomManager.markContentAsProcessed(
-    content.contentId,
-    content.room
-);
+await Promise.all([
+    this.roomManager.addMemory(
+        content.room,
+        JSON.stringify(result.content),
+        {
+            source,
+            ...result.metadata,
+            ...result.enrichedContext,
+        }
+    ),
+    this.roomManager.markContentAsProcessed(
+        content.contentId,
+        content.room
+    )
+]);

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro